package com.tcp;

import io.netty.buffer.ByteBuf;
import io.netty.util.ReferenceCountUtil;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
import lombok.Getter;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.message.codec.DefaultTransport;
import org.jetlinks.core.message.codec.EncodedMessage;
import org.jetlinks.core.message.codec.Transport;
import org.jetlinks.core.server.session.DeviceSession;
import reactor.core.publisher.Mono;

import javax.annotation.Nullable;
import java.net.SocketException;

//定义tcp session 对象
public class TcpClientSession implements DeviceSession {
    @Getter
    private  String deviceId;

    @Getter
    private DeviceOperator operator;
    private long keepALiveTimeout=-1;
    private NetSocket socket;
    private long keepAliveTimeout = System.currentTimeMillis();

    public TcpClientSession(String deviceId, DeviceOperator operator, NetSocket socket) {
        this.deviceId = deviceId;
        this.operator = operator;
        this.socket = socket;
    }

    @Override
    public String getId() {
        return "tcp_client_test_001";
    }

    @Override
    public String getDeviceId() {
        return null;
    }

    @Nullable
    @Override
    public DeviceOperator getOperator() {
        return null;
    }

    @Override
    public long lastPingTime() {
        return 0;
    }

    @Override
    public long connectTime() {
        return 0;
    }

    @Override
    public Mono<Boolean> send(EncodedMessage message) {
        return Mono
                .<Void>create((sink) -> {
                    if (socket == null) {
                        sink.error(new SocketException("socket closed"));
                        return;
                    }
                    ByteBuf buf = message.getPayload();
                    Buffer buffer = Buffer.buffer(buf);
                    socket.write(buffer, r -> {
                        if (r.succeeded()) {
                            keepAlive();
                            sink.success();
                        } else {
                            sink.error(r.cause());
                        }
                        ReferenceCountUtil.safeRelease(buf);
                    });
                })
                .thenReturn(true);
    }

    @Override
    public Transport getTransport() {
        return DefaultTransport.TCP;
    }

    @Override
    public void close() {
        socket.close();
    }

    @Override
    public void ping() {

    }

    @Override
    public boolean isAlive() {
        boolean isAlive = true;
        socket.closeHandler(close -> {
            if (isAlive) {
                socket.close();
            }
        });
        return isAlive;
    }

    @Override
    public void onClose(Runnable call) {
        call.run();
    }
}
